﻿
#ifndef __QUEUEOBJECT_H__
#define __QUEUEOBJECT_H__

#include <vector>
#include <queue>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <future>
#include <functional>
#include <chrono>
#include "noncopyable.h"

#define MAX_QUEUE_CAPACITY 1024*1024

class QueueObject : public noncopyable
{
public:
	QueueObject() :m_bStop(false), m_nCapacity(MAX_QUEUE_CAPACITY)
	{
	}

	virtual ~QueueObject()
	{
	}

	void Stop()
	{
		m_bStop.store(true);
		m_condPop.notify_all(); // 唤醒所有线程执行
	}

	//设置最大容量
	void SetMaxCapacity(int nMax)
	{
		m_nCapacity = nMax;
	}
	//获取队列任务数量
	virtual size_t GetTaskNum() = 0;


	bool IsStop()
	{
		return m_bStop;
	}

protected:
	int m_nCapacity = 0;                     //队列最大容量
	std::condition_variable_any m_condPush;  //写入条件量
	std::condition_variable_any m_condPop;   //读取条件量
	std::mutex m_mu;   //互斥锁 

	// 是否关闭提交
	std::atomic<bool> m_bStop;

};

class ThreadObject : public QueueObject
{
public:
    using QueueObject::QueueObject;
    ~ThreadObject()
    {
        Stop();
        for (std::thread& thread : m_pThreadPool) {
			if (thread.joinable())
				thread.join(); // 等待任务结束， 前提：线程一定会执行完
		}
    }
    size_t ThreadCount()
	{
		return m_pThreadPool.size();
	}
    //空闲线程数量
	int IdlCount() { return m_idlThrNum; }
protected:
    //空闲线程数量
	std::atomic<int>  m_idlThrNum ;
    std::vector<std::thread> m_pThreadPool;
};

/************************************************************************************
* 辅助函数
************************************************************************************/
//等待结果全部完成
template <typename T>
std::vector<T> wait_for_all(std::vector<std::future<T>>& vf)
{
	std::vector<T> res;
	for (auto& fu : vf)
	{
		res.push_back(fu.get());
	}
	return res;
}
 
//等待任意一个结果完成
template <typename T>
int wait_for_any(std::vector<std::future<T>>& vf, std::chrono::steady_clock::duration d)
{
	while (true)
	{
		for (int i = 0; i != vf.size(); ++i)
		{
			if (!vf[i].valid())
				continue;

			switch (vf[i].wait_for(std::chrono::seconds{0}))
			{
			case std::future_status::ready:
				return i;
			case std::future_status::timeout:
				break;
			case std::future_status::deferred:
				throw std::runtime_error("wait_for_any() deferred future");
			default:
				break;
			}
		}
		std::this_thread::sleep_for(d);
	}
}

#endif //__QUEUEOBJECT_H__